package com.xian.im.message.consumer;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.rabbitmq.client.Channel;
import com.xian.im.common.event.group.enums.GroupEventEnum;
import com.xian.im.common.event.group.model.GroupEvent;
import com.xian.im.common.utils.bussiness.ChannelClientContextUtil;
import com.xian.im.common.utils.bussiness.GroupContextUtil;
import com.xian.im.common.utils.sys.SpringContextHolder;
import com.xian.im.message.handler.SimpleMessageListenerContainerContextHolder;
import com.xian.im.message.model.QueueMsg;
import com.xian.im.message.model.RabbitMsg;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.group.ChannelGroup;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.AcknowledgeMode;
import org.springframework.amqp.core.ExchangeTypes;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.*;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.core.env.Environment;
import org.springframework.stereotype.Component;

import java.io.IOException;

/**
 * @program: xim
 * @description:
 * @author: liru.xian
 * @create: 2021-02-19 09:32
 **/
@Slf4j
@Component
public class XImSystemConsumer {

    @Autowired
    private IMCoreHandlerContext context;

    @RabbitHandler
    @RabbitListener(ackMode = "AUTO", bindings = @QueueBinding(
            value = @Queue(value = "consumer:system_" + "${xim.server.host}:${xim.server.port}", durable = "true", autoDelete = "true"),
            exchange = @Exchange(value = "consumer:system:ex", type = ExchangeTypes.FANOUT)
    ))
    public void directReceiverSystem(Message message, Channel channel) throws IOException {
        String body = new String(message.getBody());
        QueueMsg msg = JSONObject.parseObject(body, QueueMsg.class);
        method(msg);
        // 确认消费
        //channel.basicAck(message.getMessageProperties().getDeliveryTag(),true);
    }

    /**
     * 点对点
     *
     * @param message
     * @param channel
     * @throws IOException
     */
    @RabbitHandler
    @RabbitListener(ackMode = "AUTO", bindings = @QueueBinding(
            value = @Queue(value = "consumer:user_" + "${xim.server.host}:${xim.server.port}", durable = "true", autoDelete = "true"),
            exchange = @Exchange(value = "consumer:user:ex", type = ExchangeTypes.FANOUT)
    ))
    public void directReceiverUser(Message message, Channel channel) throws IOException {
        String body = new String(message.getBody());
        QueueMsg msg = JSONObject.parseObject(body, QueueMsg.class);
        method(msg);
        // 确认消费
        // channel.basicAck(message.getMessageProperties().getDeliveryTag(),true);
    }

    /**
     * 自定义消息
     *
     * @param message
     * @param channel
     * @throws IOException
     */
    @RabbitHandler
    @RabbitListener(ackMode = "AUTO", bindings = @QueueBinding(
            value = @Queue(value = "consumer:customize_" + "${xim.server.host}:${xim.server.port}", durable = "true", autoDelete = "true"),
            exchange = @Exchange(value = "consumer:customize:ex", type = ExchangeTypes.FANOUT)
    ))
    public void directReceiverCustomize(Message message, Channel channel) throws IOException {
        String body = new String(message.getBody());
        QueueMsg msg = JSONObject.parseObject(body, QueueMsg.class);
        ChannelHandlerContext ctx = ChannelClientContextUtil.getCtxById(msg.getCtxId());
        if (ctx != null) {
            method(msg);
        }
        // 确认消费
        // channel.basicAck(message.getMessageProperties().getDeliveryTag(),true);
    }

    /**
     * 群组消息发送
     *
     * @param message
     * @param channel
     * @throws IOException
     */
    @RabbitHandler
    @RabbitListener(ackMode = "AUTO", bindings = @QueueBinding(
            value = @Queue(value = "group:msg:send_" + "${xim.server.host}:${xim.server.port}", durable = "true", autoDelete = "true"),
            exchange = @Exchange(value = "group:msg:send:ex", type = ExchangeTypes.FANOUT)
    ))
    public void directReceiverGroupMsg(Message message, Channel channel) throws IOException {
        String body = new String(message.getBody());
        RabbitMsg msg = JSONObject.parseObject(body, RabbitMsg.class);
        ChannelGroup groupChannel = GroupContextUtil.getGroupChannel(msg.getId());
        groupChannel.writeAndFlush(new TextWebSocketFrame(JSON.toJSONString(msg.getResponse())));
        // 确认消费
        //channel.basicAck(message.getMessageProperties().getDeliveryTag(),true);
    }

    /**
     * 群组事件  创建  销毁群组
     *
     * @param message
     * @param channel
     * @throws IOException
     */
    @RabbitHandler
    @RabbitListener(ackMode = "MANUAL", bindings = @QueueBinding(
            value = @Queue(value = "group:event_" + "${xim.server.host}:${xim.server.port}", durable = "true", autoDelete = "true"),
            exchange = @Exchange(value = "group:event:ex", type = ExchangeTypes.FANOUT)
    ))
    public void directReceiverGroupEvent(Message message, Channel channel) throws IOException {
        String body = new String(message.getBody());
        GroupEvent msg = JSONObject.parseObject(body, GroupEvent.class);
        String groupAccount = msg.getGroupUniqueId();
        if (GroupEventEnum.CREATE.equals(msg.getEventEnum())) {
            XimGroupConsumer ximGroupConsumer = SpringContextHolder.getBean(XimGroupConsumer.class);
            SimpleMessageListenerContainer container = SpringContextHolder.getBean(SimpleMessageListenerContainer.class);
            container.addQueueNames(groupAccount + id());
            container.setAcknowledgeMode(AcknowledgeMode.MANUAL);
            container.setBatchSize(10);
            container.setConcurrentConsumers(1);
            container.setMessageListener(ximGroupConsumer);
            container.start();
            SimpleMessageListenerContainerContextHolder.put(groupAccount, container);
            log.info("群组新增消费者 {}", groupAccount);
        } else {
            SimpleMessageListenerContainer listenerContainer = SimpleMessageListenerContainerContextHolder.getAllQueue2ContainerMap().get(groupAccount);
            listenerContainer.stop();
            log.info("群组{}消费者,停止队列监听", groupAccount);
        }
        // 确认消费
        channel.basicAck(message.getMessageProperties().getDeliveryTag(), true);
    }


    public static String id() {
        Environment env = SpringContextHolder.getBean(Environment.class);
        String port = env.getProperty("xim.server.port");
        String host = env.getProperty("xim.server.host");
        return "_" + host + ":" + port;
    }


    private void method(QueueMsg msg) {
        try {
            if (msg != null) {
                context.accept(msg.getMsgRequest(), msg.getCtxId());
            }
        } catch (Exception e) {
            e.printStackTrace();
            log.debug("消息消费失败 {} 错误信息 {}", msg,  e.getMessage());
        }
    }
}
